package org.go.scheduler.database.concurrent;

import java.sql.Connection;
import java.util.HashSet;

import org.go.expcetion.LockException;
import org.go.scheduler.database.util.StdJDBCConstants;
import org.go.scheduler.database.util.TablePrefixAware;
import org.go.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Base class for database based lock handlers for providing thread/resource locking 
 * in order to protect resources from being altered by multiple threads at the 
 * same time.
 */
public abstract class DBSemaphore implements Semaphore, StdJDBCConstants, TablePrefixAware {

	private String expandedInsertSQL;

	private String expandedSQL;

	private String insertSql;
	private final Logger log = LoggerFactory.getLogger(getClass());

	private String schedName;

	private String schedNameLiteral = null;

	private String sql;
	private String tablePrefix;

	ThreadLocal<HashSet<String>> lockOwners = new ThreadLocal<HashSet<String>>();

	public DBSemaphore(String tablePrefix, String schedName, String defaultSQL, String defaultInsertSQL) {
		this.tablePrefix = tablePrefix;
		this.schedName = schedName;
		setSQL(defaultSQL);
		setInsertSQL(defaultInsertSQL);
	}

	private HashSet<String> getThreadLocks() {
		HashSet<String> threadLocks = lockOwners.get();
		if (threadLocks == null) {
			threadLocks = new HashSet<String>();
			lockOwners.set(threadLocks);
		}
		return threadLocks;
	}

	private void setExpandedSQL() {
		if (getTablePrefix() != null && getSchedName() != null && sql != null && insertSql != null) {
			expandedSQL = Util.rtp(this.sql, getTablePrefix(), getSchedulerNameLiteral());
			expandedInsertSQL = Util.rtp(this.insertSql, getTablePrefix(), getSchedulerNameLiteral());
		}
	}

	/**
	 * Execute the SQL that will lock the proper database row.
	 */
	protected abstract void executeSQL(Connection conn, String lockName, String expandedSQL, String expandedInsertSQL) throws LockException;

	protected Logger getLog() {
		return log;
	}

	protected String getSchedulerNameLiteral() {
		if (schedNameLiteral == null)
			schedNameLiteral = "'" + schedName + "'";
		return schedNameLiteral;
	}

	protected String getSQL() {
		return sql;
	}

	protected String getTablePrefix() {
		return tablePrefix;
	}

	protected void setInsertSQL(String insertSql) {
		if ((insertSql != null) && (insertSql.trim().length() != 0)) {
			this.insertSql = insertSql.trim();
		}

		setExpandedSQL();
	}

	protected void setSQL(String sql) {
		if ((sql != null) && (sql.trim().length() != 0)) {
			this.sql = sql.trim();
		}

		setExpandedSQL();
	}

	public String getSchedName() {
		return schedName;
	}

	/**
	 * Determine whether the calling thread owns a lock on the identified
	 * resource.
	 */
	@Override
	public boolean isLockOwner(Connection conn, String lockName) {
		lockName = lockName.intern();

		return getThreadLocks().contains(lockName);
	}

	/**
	 * Grants a lock on the identified resource to the calling thread (blocking
	 * until it is available).
	 * 
	 * @return true if the lock was obtained.
	 */
	@Override
	public boolean obtainLock(Connection conn, String lockName) throws LockException {

		lockName = lockName.intern();

		Logger log = getLog();

		if (log.isDebugEnabled()) {
			log.debug("Lock '" + lockName + "' is desired by: " + Thread.currentThread().getName());
		}
		if (!isLockOwner(conn, lockName)) {

			executeSQL(conn, lockName, expandedSQL, expandedInsertSQL);

			if (log.isDebugEnabled()) {
				log.debug("Lock '" + lockName + "' given to: " + Thread.currentThread().getName());
			}
			getThreadLocks().add(lockName);
			//getThreadLocksObtainer().put(lockName, new
			// Exception("Obtainer..."));
		} else if (log.isDebugEnabled()) {
			log.debug("Lock '" + lockName + "' Is already owned by: " + Thread.currentThread().getName());
		}

		return true;
	}

	/**
	 * Release the lock on the identified resource if it is held by the calling
	 * thread.
	 */
	@Override
	public void releaseLock(Connection conn, String lockName) {

		lockName = lockName.intern();

		if (isLockOwner(conn, lockName)) {
			if (getLog().isDebugEnabled()) {
				getLog().debug("Lock '" + lockName + "' returned by: " + Thread.currentThread().getName());
			}
			getThreadLocks().remove(lockName);
			//getThreadLocksObtainer().remove(lockName);
		} else if (getLog().isDebugEnabled()) {
			getLog().warn("Lock '" + lockName + "' attempt to return by: " + Thread.currentThread().getName() + " -- but not owner!", new Exception("stack-trace of wrongful returner"));
		}
	}

	/**
	 * This Semaphore implementation does use the database.
	 */
	@Override
	public boolean requiresConnection() {
		return true;
	}

	@Override
	public void setSchedName(String schedName) {
		this.schedName = schedName;

		setExpandedSQL();
	}

	@Override
	public void setTablePrefix(String tablePrefix) {
		this.tablePrefix = tablePrefix;

		setExpandedSQL();
	}
}
